package org.apache.rocketmq.client.codecoord.consumer;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.codecoord.RocketMqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 消息推送消费者，当有消息时将会回调
 *
 * @author tianxincoord@163.com
 * @date 2022/4/20
 */
public class MessagePushConsumer {
    public static void main(String[] args) throws Exception {
        // GroupName 需要 消息模 MessageModel 配合使用
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.GROUP);
        // 多个服务地址使用分号隔开，ip1:port;ip2:port;ip3:port
        consumer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDRESS);
        /// consumer.subscribe("TopicTest", "TagA || TagC || TagD");
        consumer.subscribe(RocketMqConstant.TOPIC, RocketMqConstant.TAG);
        /// 指定从哪里消费，默认从最后消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 指定消息消费模式，RocketMQ 支持两种消息模式 Clustering Broadcasting
        consumer.setMessageModel(MessageModel.BROADCASTING) ;
        // 修改消费重试次数
        consumer.setMaxReconsumeTimes(2);
        // 注册监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : msgList) {
                    System.out.println(new String(messageExt.getBody()) + " : " + JSONObject.toJSONString(context));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                // 消息重试
                // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        consumer.start();
        System.out.println("消费者启动成功");
    }
}